package sample;

import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

public class Chapter4 {
    public static void main(String[] args) throws InterruptedException {
//        Flux<Integer> ints = Flux.range(1, 3);
//        ints.subscribe();

//        Flux<Integer> ints = Flux.range(1, 3);
//        ints.subscribe(i -> System.out.println(i));
//        ints.subscribe(System.out::println);

        // 错误处理
//        Flux<Integer> ints = Flux.range(1, 4)
//                .map(i -> {
//                    if (i <= 3) return i;
//                    throw new RuntimeException("Got to 4");
//                });
//        ints.subscribe(i -> System.out.println(i),
//                error -> System.err.println("Error: " + error));

        // 既有错误处理，还有一个完成后的处理
        // 错误和完成信号都是终止信号，并且二者只会出现其中之一。为了能够最终全部正常完成，你必须处理错误信号。
//        Flux<Integer> ints = Flux.range(1, 4);
//        ints.subscribe(i -> System.out.println(i),
//                error -> System.err.println("Error " + error),
//                () -> {System.out.println("Done");});

//        SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
//        Flux<Integer> ints = Flux.range(1, 4);
//        ints.subscribe(i -> System.out.println(i),
//                error -> System.err.println("Error " + error),
//                () -> {System.out.println("Done");},
//                s -> ss.request(10));
//        ints.subscribe(ss);

//        Flux<String> source = someStringSource();
//
//        source.map(String::toUpperCase)
//                .subscribe(new BaseSubscriber<String>() {
//                    @Override
//                    protected void hookOnSubscribe(Subscription subscription) {
//
//                        request(1);
//                    }
//
//                    @Override
//                    protected void hookOnNext(String value) {
//                        request(1);
//                    }
//
//
//                });
        /*
        BaseSubscriber 是一个抽象类，所以我们创建一个匿名内部类。
        BaseSubscriber 定义了多种用于处理不同信号的 hook。它还定义了一些捕获 Subscription 对象的现成方法，这些方法可以用在 hook 中。
        request(n) 就是这样一个方法。它能够在任何 hook 中，通过 subscription 向上游传递 背压请求。这里我们在开始这个流的时候请求1个元素值。
        随着接收到新的值，我们继续以每次请求一个元素的节奏从源头请求值。
        其他 hooks 有 hookOnComplete, hookOnError, hookOnCancel, and hookFinally （它会在流终止的时候被调用，传入一个 SignalType 作为参数）。
        当你修改请求操作的时候，你必须注意让 subscriber 向上提出足够的需求， 否则上游的 Flux 可能会被“卡住”。
        所以 BaseSubscriber 在进行扩展的时候要覆盖 hookOnSubscribe 和 onNext，这样你至少会调用 request 一次。
        BaseSubscriber 还提供了 requestUnbounded() 方法来切换到“无限”模式（等同于 request(Long.MAX_VALUE)）。
         */
//        Flux<String> flux = Flux.generate(
//                () -> 0,
//                (state, sink) -> {
//                    sink.next("3 x " + state + " = " + 3*state);
//                    if (state == 10) sink.complete();
//                    return state + 1;
//                });
        /*
        初始化状态值（state）为0。
        我们基于状态值 state 来生成下一个值（state 乘以 3）。
        我们也可以用状态值来决定什么时候终止序列。
        返回一个新的状态值 state，用于下一次调用。
         */
//        Flux<String> flux1 = Flux.generate(
//                AtomicLong::new,
//                (state, sink) -> {
//                    long i = state.getAndIncrement();
//                    sink.next("3 x " + i + " = " + 3*i);
//                    if (i == 10) sink.complete();
//                    return state;
//                });
        /*
        这次我们初始化一个可变类型的状态值。
        改变状态值。
        返回 同一个 实例作为新的状态值。
         */
//        Flux<String> flux2 = Flux.generate(
//                AtomicLong::new,
//                (state, sink) -> {
//                    long i = state.getAndIncrement();
//                    sink.next("3 x " + i + " = " + 3*i);
//                    if (i == 10) sink.complete();
//                    return state;
//                }, (state) -> System.out.println("state: " + state));
//
//        Flux.range(1, 10000)
//                .publishOn(Schedulers.parallel())
//                .subscribe(result);

//        Flux<String> s = Flux.range(1, 10)
//                .map(v -> doSomethingDangerous(v))
//                .map(v -> doSecondTransform(v));
//        s.subscribe(value -> System.out.println("RECEIVED " + value),
//                error -> System.err.println("CAUGHT " + error)
//        );
//        try {
//            for (int i = 1; i < 11; i++) {
//                String v1 = doSomethingDangerous(i);
//                String v2 = doSecondTransform(v1);
//                System.out.println("RECEIVED " + v2);
//            }
//        } catch (Throwable t) {
//            System.err.println("CAUGHT " + t);
//        }

//        Flux.just(10)
//                .map(this::doSomethingDangerous)
//                .onErrorReturn("RECOVERED");
//        Flux.just(10)
//                .map(this::doSomethingDangerous)
//                .onErrorReturn(e -> e.getMessage().equals("boom10"), "recovered10");
//        Flux.just("key1", "key2")
//                .flatMap(k -> callExternalService(k))
//                .onErrorResume(e -> getFromCache(k));
//        Flux.just("timeout1", "unknown", "key2")
//                .flatMap(k -> callExternalService(k))
//                .onErrorResume(error -> {
//                    if (error instanceof TimeoutException)
//                        return getFromCache(k);
//                    else if (error instanceof UnknownKeyException)
//                        return registerNewEntry(k, "DEFAULT");
//                    else
//                        return Flux.error(error);
//                });
//        erroringFlux.onErrorResume(error -> Mono.just(
//                myWrapper.fromError(error)
//        ));
//        Flux.just("timeout1")
//                .flatMap(k -> callExternalService(k))
//                .onErrorResume(original -> Flux.error(
//                        new BusinessException("oops, SLA exceeded", original)
//                );
//        Flux.just("timeout1")
//                .flatMap(k -> callExternalService(k))
//                .onErrorMap(original -> new BusinessException("oops, SLA exceeded", original));

//        LongAdder failureStat = new LongAdder();
//        Flux<String> flux =
//                Flux.just("unknown")
//                        .flatMap(k -> callExternalService(k))
//                        .doOnError(e -> {
//                            failureStat.increment();
//                            log("uh oh, falling back, service failed for key " + k);
//                        })
//                        .onErrorResume(e -> getFromCache(k));

//        AtomicBoolean isDisposed = new AtomicBoolean();
//        Disposable disposableInstance = new Disposable() {
//            @Override
//            public void dispose() {
//                isDisposed.set(true);
//            }
//
//            @Override
//            public String toString() {
//                return "DISPOSABLE";
//            }
//        };
//
//        Flux<String> flux =
//                Flux.using(
//                        () -> disposableInstance,
//                        disposable -> Flux.just(disposable.toString()),
//                        Disposable::dispose
//                );

//        LongAdder statsCancel = new LongAdder();
//
//        Flux<String> flux =
//                Flux.just("foo", "bar")
//                        .doFinally(type -> {
//                            if (type == SignalType.CANCEL)
//                                statsCancel.increment();
//                        })
//                        .take(1);

//        Flux<String> flux =
//                Flux.interval(Duration.ofMillis(250))
//                        .map(input -> {
//                            if (input < 3) return "tick " + input;
//                            throw new RuntimeException("boom");
//                        })
//                        .onErrorReturn("Uh oh");
//
//        flux.subscribe(System.out::println);
//        Thread.sleep(2100);
        // 注意 interval 默认基于一个 timer Scheduler 来执行。 如果我们想在 main 方法中运行，
        // 我们需要调用 sleep，这样程序就可以在还没有产生任何值的时候就退出了。
//        Flux.interval(Duration.ofMillis(250))
//                .map(input -> {
//                    if (input < 3) return "--tick " + input;
//                    throw new RuntimeException("boom");
//                })
//                .elapsed()
//                .retry(2)
////                .retry(1)
//                .subscribe(System.out::println, System.err::println);
//
//        Thread.sleep(4100);
////        Thread.sleep(2100);
        /*
        elapsed 会关联从当前值与上个值发出的时间间隔（译者加：如下边输出的内容中的 259/249/251…​）。
        我们还是要看一下 onError 时的内容。
        确保我们有足够的时间可以进行 4x2 次 tick。
         */
//        Flux<String> flux = Flux
//                .<String>error(new IllegalArgumentException())
//                .doOnError(System.out::println)
//                .retryWhen(companion -> companion.take(3));
//        Thread.sleep(2100);
        /*
        持续产生错误。
        在 retry 之前 的 doOnError 可以让我们看到错误。
        这里，我们认为前 3 个错误是可以重试的（take(3)），再有错误就放弃。
        事实上，上边例子最终得到的是一个 空的 Flux，但是却 成功 完成了。反观对同一个 Flux 调用 retry(3) 的话，
        最终是以最后一个 error 终止 Flux，故而 retryWhen 与之不同。
         */
//        Flux<String> flux =
//            Flux.<String>error(new IllegalArgumentException())
//                .retryWhen(companion -> companion
//                        .zipWith(Flux.range(1, 4),
//                                (error, index) -> {
//                                    if (index < 4) return index;
//                                    else throw Exceptions.propagate(error);
//                                })
//                );
//        Thread.sleep(2100);
        /*
        技巧一：使用 zip 和一个“重试个数 + 1”的 range。
        zip 方法让你可以在对重试次数计数的同时，仍掌握着原始的错误（error）。
        允许三次重试，小于 4 的时候发出一个值。
        为了使序列以错误结束。我们将原始异常在三次重试之后抛出。
         */
//        Flux.just("foo")
//            .map(s -> { throw new IllegalArgumentException(s); })
//            .subscribe(v -> System.out.println("GOT VALUE"),
//                    e -> System.out.println("ERROR: " + e));

        Flux<String> converted = Flux
            .range(1, 10)
            .map(i -> {
                try { return convert(i); }
                catch (IOException e) { throw Exceptions.propagate(e); }
            });
        converted.subscribe(
            v -> System.out.println("RECEIVED: " + v),
            e -> {
                if (Exceptions.unwrap(e) instanceof IOException) {
                    System.out.println("Something bad happened with I/O");
                } else {
                    System.out.println("Something bad happened");
                }
            }
        );

        /*
        创建一个有 10,000 个元素的 Flux。
        创建等同于 CPU 个数的线程（最小为4）。
        subscribe() 之前什么都不会发生。
        */

        //    Flux<String> bridge = Flux.create(sink -> {
//        myEventProcessor.register(
//                new MyEventListener<String>() {
//
//                    public void onDataChunk(List<String> chunk) {
//                        for(String s : chunk) {
//                            sink.next(s);
//                        }
//                    }
//
//                    public void processComplete() {
//                        sink.complete();
//                    }
//                });
//    });
    /*
    	桥接 MyEventListener。
        每一个 chunk 的数据转化为 Flux 中的一个元素。
        processComplete 事件转换为 onComplete。
        所有这些都是在 myEventProcessor 执行时异步执行的。
     */

//    Flux<String> bridge = Flux.push(sink -> {
//        myEventProcessor.register(
//                new SingleThreadEventListener<String>() {
//
//                    public void onDataChunk(List<String> chunk) {
//                        for(String s : chunk) {
//                            sink.next(s);
//                        }
//                    }
//
//                    public void processComplete() {
//                        sink.complete();
//                    }
//
//                    public void processError(Throwable e) {
//                        sink.error(e);
//                    }
//                });
//    });
    /*
    桥接 SingleThreadEventListener API。
    在监听器所在线程中，事件通过调用 next 被推送到 sink。
    complete 事件也在同一个线程中。
    error 事件也在同一个线程中。
     */

//    Flux<String> bridge = Flux.create(sink -> {
//        myMessageProcessor.register(
//                new MyMessageListener<String>() {
//
//                    public void onMessage(List<String> messages) {
//                        for(String s : messages) {
//                            sink.next(s);
//                        }
//                    }
//                });
//        sink.onRequest(n -> {
//            List<String> messages = myMessageProcessor.request(n);
//            for(String s : message) {
//                sink.next(s);
//            }
//        });
    /*
    当有请求的时候取出一个 message。
    如果有就绪的 message，就发送到 sink。
    后续异步到达的 message 也会被发送给 sink。
     */

//    Flux<String> bridge = Flux.create(sink -> {
//        sink.onRequest(n -> channel.poll(n))
//                .onCancel(() -> channel.cancel()) // 在取消时被调用。
//                .onDispose(() -> channel.close()) // 在有完成、错误和取消时被调用。
//    });

//    public String alphabet(int letterNumber) {
//        if (letterNumber < 1 || letterNumber > 26) {
//            return null;
//        }
//        int letterIndexAscii = 'A' + letterNumber - 1;
//        return "" + (char) letterIndexAscii;
//    }
//
//    Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20)
//            .handle((i, sink) -> {
//                String letter = alphabet(i);
//                if (letter != null)
//                    sink.next(letter);
//            });
//
//    alphabet.subscribe(System.out::println);
    }

    public static String convert(int i) throws IOException {
        if (i > 3) {
            throw new IOException("boom " + i);
        }
        return "OK " + i;
    }
}
